package com.tron.sync.transferElkData;

import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.elasticsearch.client.ClientConfiguration;
import org.springframework.data.elasticsearch.client.RestClients;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchScrollHits;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;

@Service
public class TransferElkDataServer {

    @Value("${fromElk}")
    private String fromElk;
    @Value("${fromUsername}")
    private String fromUsername;
    @Value("${fromPassword}")
    private String fromPassword;
    @Value("${toElk}")
    private String toElk;
    @Value("${toUsername}")
    private String toUsername;
    @Value("${toPassword}")
    private String toPassword;

    public Object transferWfinfoElkData() {
        final ElasticsearchRestTemplate from = getElasticsearchRestTemplate(fromElk.split(","), fromUsername, fromPassword);
        final ElasticsearchRestTemplate to = getElasticsearchRestTemplate(toElk.split(","), toUsername, toPassword);
        final ArrayList<WfinfoData> WfinfoDatas = new ArrayList<>();
        final NativeSearchQuery build = new NativeSearchQueryBuilder().withQuery(QueryBuilders.matchAllQuery()).build();
        build.setMaxResults(1000);
        final SearchScrollHits<WfinfoData> searchHits = from.searchScrollStart(5000, build, WfinfoData.class, IndexCoordinates.of("wfinfoindex"));
        final Stream<SearchHit<WfinfoData>> hitStream = searchHits.get();
        hitStream.forEach(e->{
            WfinfoDatas.add(e.getContent());
        });
        to.save(WfinfoDatas);
        WfinfoDatas.clear();
        // 最大条数
        long totalHits = searchHits.getTotalHits();
        // 滚动id
        final String scrollId = searchHits.getScrollId();
        // count
        int count = 0;


        while (totalHits != 0) {
            System.out.println("滚动次数："+count++);
            final SearchScrollHits<WfinfoData> user = from.searchScrollContinue(scrollId, 5000, WfinfoData.class, IndexCoordinates.of("wfinfoindex"));
            final List<SearchHit<WfinfoData>> list = user.getSearchHits();
            if (list.size() == 0) {
                break;
            }else {
                final Stream<SearchHit<WfinfoData>> stream = list.stream();
                stream.forEach(e->{
                    WfinfoDatas.add(e.getContent());
                });
                to.save(WfinfoDatas);
                WfinfoDatas.clear();
            }
        }
        from.searchScrollClear(Arrays.asList(scrollId));
        return totalHits;
    }

    public ElasticsearchRestTemplate getElasticsearchRestTemplate(String[] hostAndPorts,String username,String password) {
        final ElasticsearchRestTemplate elasticsearchRestTemplate = new ElasticsearchRestTemplate(elasticsearchClient(hostAndPorts,username,password));
        return elasticsearchRestTemplate;
    }

    private RestHighLevelClient elasticsearchClient(String[] hostAndPorts,String username,String password) {
        final ClientConfiguration clientConfiguration = ClientConfiguration.builder()
                .connectedTo(hostAndPorts)
                .withBasicAuth(username, password)
                .build();
        return RestClients.create(clientConfiguration).rest();
    }


    public String syncData(String endTime){
        ElasticsearchRestTemplate from = getElasticsearchRestTemplate(fromElk.split(","), fromUsername, fromPassword);
        ElasticsearchRestTemplate to = getElasticsearchRestTemplate(toElk.split(","), toUsername, toPassword);
        NativeSearchQuery build = new NativeSearchQueryBuilder().withQuery(QueryBuilders.rangeQuery("createTime").lte(endTime)).build();
        build.setMaxResults(1000);
        final SearchScrollHits<WfinfoData> searchHits = from.searchScrollStart(5000, build, WfinfoData.class, IndexCoordinates.of("wfinfoindex"));
        final Stream<SearchHit<WfinfoData>> hitStream = searchHits.get();
        return "";
    }




}
